# 这三行是固定的
from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("实例1").master("local[*]").getOrCreate()
sc = spark.sparkContext

# 键值对rdd
# textFile = sc.textFile("file////E:/PyCharm/PythonFolder/json/maprdd.txt")
# # map 将所有的文件数据，经过map里面的函数进行计算或组合
# # 在这里，就是将w（原数据）变成(w，1)这样的键值对
# # 如，原本是111，现在变成(111,1)。
# # 用处就是，111为key，1位值（有1个111）可以统计到111有多少个
# pairRdd = textFile.map(lambda w: (w, 1))
# pairRdd.foreach(print)

# 通过并行集合创建RDD
# list = ['xiaomi', 'huawei', 'sanxing', 'oppo', 'xiaomi', 'xiaomi']
# # parallelize可以并行处理数据，
# # list后面可以加一个数字，表示并行量
# rdd = sc.parallelize(list)
# pairRdd = rdd.map(lambda w: (w, 1))
# pairRdd.foreach(print)

# reduceByKey（统计重复的数量）
# list = ['xiaomi', 'huawei', 'sanxing', 'oppo', 'xiaomi', 'xiaomi']
# rdd = sc.parallelize(list)
# pairRdd = rdd.map(lambda w: (w, 1))
# # 相同的key进行相加
# reduce = pairRdd.reduceByKey(lambda a, b: a + b)
# # 将reduce转变成数组
# list = reduce.collect()
# for l in list:
#     print(l)

# 单独获得key或value
# list = [
#     "peach", "plum", "banana", "banana", "plum", "plum", "peach", "peach",
#     "peach"
# ]
# rdd = sc.parallelize(list)
# pairRDD = rdd.map(lambda word: (word, 1))
# # 统计的结果
# newRdd = pairRDD.reduceByKey(lambda a, b: a + b)
# # 得到单独的key（需要转换成列表）
# keyRdd = newRdd.keys()
# tmp_list = keyRdd.collect()
# print("keys=", tmp_list)
# # 得到单独的value（需要转换成列表）
# keyRdd = newRdd.values()
# tmp_list = keyRdd.collect()
# print("values=", tmp_list)

# sortByKey 根据key来排序（用的比较少）
# list = ["peach", "plum", "banana", "banana",  "plum", "plum", "peach", "peach", "peach","apple"]
# rdd = sc.parallelize(list)
# pairRDD = rdd.map(lambda word: (word, 1))
# newRdd = pairRDD.reduceByKey(lambda a, b: a + b)
# keyRdd = newRdd.sortByKey()
# tmp_list = keyRdd.collect()
# print(tmp_list)

# sortBy 可以自己设置按什么方法进行排序
list = [
    "peach", "plum", "banana", "banana", "plum", "plum", "peach", "peach",
    "peach", "apple"
]
rdd = sc.parallelize(list)
pairRDD = rdd.map(lambda word: (word, 1))
newRdd = pairRDD.reduceByKey(lambda a, b: a + b)
# 按照values降序（取x[1]即values）
# ascending=True，从小到大
# ascending=False，从大到小
keyRdd = newRdd.sortBy(lambda x: x[1], ascending=False)
tmp_list = keyRdd.collect()
print(tmp_list)

# mapValues 对values进行修改（比如将0.01变成1%等）

# join 将两个数组与数据库一样，进行连接起来
# join里面包括了join,leftOuterJoin,rightOuterJoin,fullOuterJoin
